use mobc::Pool;
use mobc_postgres::PgConnectionManager;
use std::str::FromStr;

use tokio_postgres::types::ToSql;

//use postgres_types::{ToSql, FromSql};
use tokio_postgres::Config;
use tokio_postgres::NoTls;

use futures::{pin_mut, TryStreamExt};
//use futures::executor::block_on;
//初始化连接池对象
lazy_static! {
    pub static ref PSQLPOOL: mobc::Pool< PgConnectionManager<NoTls>> = {
        println!("lazy_static: 初始化PostgresSQL数据库连接池...");
        //block_on(psql_pool())
        psql_pool()
    };
}

// 创建连接池
// 如果实现异步需要在执行者上运行才能起作用
// pub async fn psql_pool() ->Pool<PgConnectionManager<NoTls>>
pub fn psql_pool() -> Pool<PgConnectionManager<NoTls>> {
    let config = Config::from_str("postgres://postgres@localhost:5432/lin").unwrap();
    let manager = PgConnectionManager::new(config, NoTls);
    let pool = Pool::builder().max_open(20).build(manager);
    pool
    // const MAX: usize = 5000;
    // let now = Instant::now();
    // let (tx, mut rx) = tokio::sync::mpsc::channel::<usize>(16);
    // for i in 0..MA X {
    //     let pool = pool.clone();
    //     let mut tx_c = tx.clone();
    //     tokio::spawn(async move {
    //         let client = pool.get().await.unwrap();
    //         let rows = client.query("SELECT 1 + 2", &[]).await.unwrap();
    //         let value: i32 = rows[0].get(0);
    //         assert_eq!(value, 3);
    //         tx_c.send(i).await.unwrap();
    //     });
    // }
    // for _ in 0..MAX {
    //     rx.recv().await.unwrap();
    // }
    // println!("cost: {:?}", now.elapsed());
}

// 测试连接并初始化
pub async fn ping_psql() {
    // 首次调用初始化了lazy_static
    let pool = PSQLPOOL.clone();
    let client = pool.get().await.unwrap();
    let rows = client.query("SELECT 1+2;", &[]).await.unwrap();
    let value: i32 = rows[0].get(0);
    assert_eq!(value, 3);
}

pub struct PsqlExec<'a> {
    // conn 是连接管理
    conn: &'a Pool<PgConnectionManager<NoTls>>,
}

// #[derive(Debug, ToSql, FromSql)]
// enum Mood {
//     Sad,
//     Ok,
//     Happy,
// }
impl<'a> PsqlExec<'a> {
    // 创建PsqlExec对象
    pub fn new(pool: &'a Pool<PgConnectionManager<NoTls>>) -> Self {
        PsqlExec { conn: pool }
    }

    // hexinfo 表插入数据
    pub fn inster_hexinfo(&self, hex: String, info: String, nodeid: String, chunk_id: i32) {
        //use chrono::prelude::Local;
        let pool = self.conn.clone();
        tokio::spawn(async move {
            let client = pool.get().await.unwrap();
            // let params: Vec<String> = vec![hex, info, nodeid,chunk_id];
            match client
                .query(
                    "INSERT INTO HEXINFO 
            (HEX,SPAWNTIME,INFO,NODEID,CHUNKID) 
            VALUES($1,current_timestamp,$2,$3,$4);",
                    &[&hex, &info, &nodeid, &chunk_id],
                )
                .await
            {
                Ok(it) => it,
                Err(err) => panic!("query_raw hexinfo: {}", err),
            };
            // pin_mut!(it);
            // match it.try_next().await {
            //     Ok(row) => row,
            //     Err(err) => panic!("try_next() : {}", err),
            // };

            //println!("inster hexinfo black info :{:?}", row)
        });
    }

    // hexinfo 表读取
    pub fn get_row_hexinfo<F>(&self, func: F, id: String)
    where
        F: Fn(String, std::time::SystemTime, String, String) + Send + Sync + 'static,
    {
        let pool = self.conn.clone();
        tokio::spawn(async move {
            //let params: Vec<String> = vec![id];
            //let s  = String::from("xxxxxx");
            //let mut s_out;
            //ToSql::to_sql(s, s_out);
            let client = pool.get().await.unwrap();
            //let params: Vec<String> = vec![hex, info, nodeid];
            let row = match client
                .query_one(
                    "SELECT * FROM HEXINFO WHERE HEX = $1",
                    &[&id],
                    //params.iter().map(|p| p as &dyn ToSql),
                )
                .await
            {
                Ok(it) => it,
                Err(err) => panic!("query_raw hexinfo: {}", err),
            };
            let hex: String = row.get(0);
            let time: std::time::SystemTime = row.get(1);
            let info: String = row.get(2);
            let nodeid: String = row.get(3);
            func(hex, time, info, nodeid)
        });
    }

    // hexinfo 表读取多个
    pub fn get_rows_hexinfo<F>(&self, func: F)
    where
        F: Fn(Vec<tokio_postgres::row::Row>) + Send + Sync + 'static,
    {
        let pool = self.conn.clone();
        tokio::spawn(async move {
            //let params: Vec<String> = vec![id];
            //let s  = String::from("xxxxxx");
            //let mut s_out;
            //ToSql::to_sql(s, s_out);
            let client = pool.get().await.unwrap();
            //let params: Vec<String> = vec![hex, info, nodeid];
            let rows = match client
                .query(
                    "SELECT * FROM HEXINFO  ",
                    &[], //params.iter().map(|p| p as &dyn ToSql),
                )
                .await
            {
                Ok(it) => it,
                Err(err) => panic!("query_raw hexinfo: {}", err),
            };
            func(rows)
        });
    }

    // 测试曲query_raw

    pub fn my_query_raw<F>(&self, table_name: String, col_name: String, value: String, func: F)
    where
        F: Fn(tokio_postgres::row::Row) + Send + Sync + 'static,
    {
        let pool = self.conn.clone();
        tokio::spawn(async move {
            //let params: Vec<String> = vec![id];
            let params: Vec<String> = vec![value];
            //let s = s.to_sql(s,s_out);
            //let mut s_out;
            //ToSql::to_sql(s, s_out);
            let client = &pool.get().await.unwrap() as &tokio_postgres::Client;
            let s: String = String::from("SELECT ")
                + &col_name
                + " FROM "
                + &table_name
                + " WHERE "
                + &col_name
                + " = $1";

            //let s = &s as &dyn ToSql;
            let mut it = match client
                .query_raw(
                    s.as_str(),
                    params.iter().map(|p| p as &dyn ToSql), //params.iter().map(|p| p as &dyn ToSql),
                )
                .await
            {
                Ok(it) => it,
                Err(err) => panic!("query_raw hexinfo: {}", err),
            };
            pin_mut!(it);
            while let Some(row) = it.try_next().await.unwrap() {
                func(row);
            }
        });
    }
}
